﻿// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. 
// See License.txt in the project root for license information.
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace System.Threading.Actors.Utilities
{
    // Specialized queue that allows for synchronous queuing and asynchronous dequeuing.
    public class AsyncQueue<T>
    {
        // Actual queue that stores items
        private readonly Queue<T> m_itemQueue = new Queue<T>();

        // A queue of TaskCompletionSources used to wake consumers waiting for input
        private readonly Queue<TaskCompletionSource<T>> m_waiterQueue = new Queue<TaskCompletionSource<T>>();

        // An async mutex to protect access to the internals of this class
        private readonly SemaphoreSlim m_asyncMutex = new SemaphoreSlim(1, 1);

        // An indication that no more adding will be taking place
        private bool m_addingCompleted = false;

        public int Count
        {
            get
            {
                // Is there any need to lock around this?
                return m_itemQueue.Count;
            }
        }

        /// <summary>
        /// Enqueue an item, synchronously. 
        /// </summary>
        /// <param name="item">The item to be enqueued</param>
        /// <returns></returns>
        public void Enqueue(T item, bool underLock = false)
        {
            TaskCompletionSource<T> released_tcs = null;

            m_asyncMutex.Wait(); // gain exclusive access to internals
            if (m_addingCompleted)
            {
                // CompleteAddingAsync has been called, so this operation is illegal.
                m_asyncMutex.Release();
                throw new InvalidOperationException("Attempt to Add when queue has been marked as complete for adding");
            }

            try
            {
                if (m_waiterQueue.Count > 0)
                {
                    Debug.Assert(m_itemQueue.Count == 0, "Expected item queue to be empty since there were waiters");
                    // There are waiters.  Wake one directly, without putting the item into the queue
                    released_tcs = m_waiterQueue.Dequeue();
                }
                else
                {
                    // No waiters.  Store the item in the queue.
                    m_itemQueue.Enqueue(item);
                }
            }
            finally
            {
                m_asyncMutex.Release(); // release exclusive access to internals
            }

            // If we need to release a waiter, do so outside the m_asyncMutex lock.
            if (released_tcs != null)
            {
                if (underLock)
                {
                    // If the producer is under lock, we'll need to release the waiter on a separate task/thread in
                    // order to avoid an await waking up under lock.
                    Task.Factory.StartNew(tupleAsObject =>
                    {
                        var tuple = (Tuple<T, TaskCompletionSource<T>>)tupleAsObject;
                        tuple.Item2.TrySetResult(tuple.Item1);
                    }, Tuple.Create(item, released_tcs), CancellationToken.None, TaskCreationOptions.PreferFairness, TaskScheduler.Default);
                }
                else
                {
                    released_tcs.TrySetResult(item);
                }
            }
        }

        /// <summary>
        /// Mark the AsyncQueue as "complete for adding".  This will cause any ensuing enqueue operations
        /// to throw an InvalidOperationException, and will immediately wake any waiting consumers and
        /// cause them to fail with an EndOfQueueException.
        /// </summary>
        /// <param name="underLock">If true, then any waiter-waking will be done from a separate task in
        /// order to avoid awaits waking under lock.</param>
        /// <returns></returns>
        public void CompleteAdding(bool underLock=false)
        {
            m_asyncMutex.Wait(); // gain exclusive access to internals
            m_addingCompleted = true;

            List<TaskCompletionSource<T>> toFaultList = null;
            try
            {
                // If there are waiters remaining, mark them for faulting
                if (m_waiterQueue.Count > 0)
                {
                    toFaultList = new List<TaskCompletionSource<T>>(m_waiterQueue);
                    m_waiterQueue.Clear();
                }
            }
            finally
            {
                m_asyncMutex.Release(); // release exclusive access to internals                    
            }

            // Fault the toFaultList once you leave the lock
            if (toFaultList != null)
            {
                if (underLock)
                {
                    // If we are under lock, then we need to wake consumers from a background task in order
                    // to avoid them awaking under lock.
                    Task.Factory.StartNew(listAsObject =>
                    {
                        var faultList = (List<TaskCompletionSource<T>>)listAsObject;
                        var e = new EndOfQueueException("No new elements will be added to this AsyncQueue.");
                        foreach (var tcs in faultList) tcs.TrySetException(e);
                    }, toFaultList, CancellationToken.None, TaskCreationOptions.PreferFairness, TaskScheduler.Default);
                }
                else
                {
                    var e = new EndOfQueueException("No new elements will be added to this AsyncQueue.");
                    foreach (var tcs in toFaultList) tcs.TrySetException(e);
                }
            }

        }

        /// <summary>
        /// Dequeues an item from the queue asynchronously.  If an item is available, it will
        /// be dequeued and returned.  If not, then an entry for this operation will be created 
        /// on the wait list.
        /// </summary>
        /// <returns></returns>
        public Task<T> DequeueAsync()
        {
            m_asyncMutex.Wait(); // gain exclusive access to internals

            if (m_itemQueue.Count > 0)
            {
                // We can immediately return a value.  Do so.
                var result = m_itemQueue.Dequeue();
                m_asyncMutex.Release();
                return Task.FromResult(result);
            }

            // No value available right now.  
            if (m_addingCompleted)
            {
                // No more new data is on the way.
                // Release the lock and throw an exception
                m_asyncMutex.Release();
                throw new EndOfQueueException("No new elements will be added to this AsyncQueue.");
            }

            //Queue yourself as a waiter.
            var tcs = new TaskCompletionSource<T>();
            m_waiterQueue.Enqueue(tcs);
            m_asyncMutex.Release();

            return tcs.Task;
        }



        /// <summary>
        /// Dequeues all available items from the queue.  
        /// </summary>
        /// <returns></returns>
        public Task<IEnumerable<T>> DequeueAllAsync()
        {
            m_asyncMutex.Wait(); // gain exclusive access to internals

            if (m_itemQueue.Count > 0)
            {
                // We have one or more items.  Return them to the requestor.
                var results = new List<T>(m_itemQueue);
                m_itemQueue.Clear();
                m_asyncMutex.Release();
                return Task.FromResult((IEnumerable<T>)results);
            }

            // No value available right now.  
            if (m_addingCompleted)
            {
                // No more new data is on the way.
                // Release the lock and throw an exception
                m_asyncMutex.Release();
                throw new EndOfQueueException("No new elements will be added to this AsyncQueue.");
            }

            // Queue yourself as a waiter.
            var tcs = new TaskCompletionSource<T>();
            m_waiterQueue.Enqueue(tcs);
            m_asyncMutex.Release();

            // Wait for a single value, return it in the form of a one-element list
            return tcs.Task.ContinueWith(antecedent =>
            {
                var returnList = new List<T>();
                returnList.Add(antecedent.Result);
                return (IEnumerable<T>)returnList;
            });
        }
    }



}